package com.wxx.socket.server;

import com.wxx.socket.BroadcastUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.concurrent.GlobalEventExecutor;

/**
 * webSocket数据处理
 *
 * @author 万祥新
 *
 */
public class WebSocketChannelHandler extends SimpleChannelInboundHandler<Object> {
    String uri;

    public WebSocketChannelHandler(String uri) {
        this.uri = uri;
    }

    WebSocketServerHandshaker handshaker;
    public static ChannelGroup webSocketGroup = new DefaultChannelGroup("websocket", GlobalEventExecutor.INSTANCE);

    @Override
    protected void channelRead0(ChannelHandlerContext arg0, Object msg) {
        if (msg instanceof FullHttpRequest) {
            handlerHttpRequest(arg0, (FullHttpRequest) msg);
        } else if (msg instanceof WebSocketFrame) {
            handlerWebsocketFrame(arg0, (WebSocketFrame) msg);
        }
    }

    private void handlerHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {
        if (!req.getDecoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))) {

            return;
        }

        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(uri, null, false);
        handshaker = wsFactory.newHandshaker(req);
        if (handshaker == null) {
            WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
        } else {
            System.out.println("WebSocket新用户连接了");
            // 将http协议升级为socket协议
            handshaker.handshake(ctx.channel(), req);
            webSocketGroup.add(ctx.channel());
            BroadcastUtil.broadcastToAll("WebSocket新用户连接了");
        }
    }

    private void handlerWebsocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
        if (frame instanceof CloseWebSocketFrame) {
            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
            System.out.println("WebSocket用户退出了");
            webSocketGroup.remove(ctx.channel());
            BroadcastUtil.broadcastToAll("WebSocket用户退出了");
            return;
        }

        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }

        if (!(frame instanceof TextWebSocketFrame)) {
            System.out.println("不支持的协议");
            return;
        }

        String content = ((TextWebSocketFrame) frame).text();
        System.out.println("WebSocket收到数据>>>" + content);
        BroadcastUtil.broadcastToAll(content);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
        Channel channel = ctx.channel();
        webSocketGroup.remove(channel);
        if (channel.isActive()) {
            ctx.channel();
        }
    }
}